package regular;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;

import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;

public class EvalMapper extends Mapper<Text, Text, NullWritable, RvecWritable> {

	public void map(Text key, Text value, Context context) throws IOException,
			InterruptedException {

		Initial.COUNTER += 1;
		Initial.CONF = context.getConfiguration();

		// get query
		Path[] fileCaches = DistributedCache.getLocalCacheFiles(context
				.getConfiguration());
		GraphQuery gq = Ultility.getAutomaton(fileCaches[0]);
		// System.out.print(gq.toString());

		// get fragment info
		HashMap<String, Vertex> listViNode = new HashMap<String, Vertex>();
		HashMap<String, Vertex> listInNode = new HashMap<String, Vertex>();
		HashMap<String, Vertex> listOutNode = new HashMap<String, Vertex>();

		long startTime1 = System.currentTimeMillis();
		LoadDataYouTube(value, listViNode, listInNode, listOutNode);
		// LoadDataDBLP(value, listViNode, listInNode, listOutNode);
		// for(Vertex v: listOutNode.values())
		// System.out.println(v.name);
		long endTime1 = System.currentTimeMillis();
		HashSet<Rvec> rvset = localEval(listViNode, listInNode, listOutNode, gq);

		long endTime2 = System.currentTimeMillis();
		RvecWritable values = new RvecWritable();
		values.rvset = rvset;

		int mapId = Integer.parseInt(context.getTaskAttemptID().toString()
				.split("_")[4]);
		// if(mapId % 2==0)
		// k = new IntWritable(1);
		// else
		// k = new IntWritable(2);

		// multipleOutputs.write(NullWritable.get(), values, k.toString());

		HashSet<String> listOut = new HashSet<String>();
		StringBuilder sbIn = new StringBuilder();
		StringBuilder sb = new StringBuilder();
		for (Rvec r : rvset) {
			for (BoolFormula b : r.value) {
				if (!listOut.contains(b.toString())) {
					if (!b.value) {
						listOut.add(b.toString());
						sb.append(b.toString() + "\n");
					} else {
						if (!listOut.contains(gq.ut.stateName)) {
							listOut.add(gq.ut.stateName);
							sb.append(gq.ut.stateName + "\n");
						}
					}
				}
			}
			sbIn.append(r.v + "\n");
		}

		FileSystem fs = FileSystem.get(Initial.CONF);
		Path pathOutputNode = new Path(Initial.PATH_OUTPUT_NODE_CACHE + "/"
				+ mapId);
		BufferedWriter bwOut = new BufferedWriter(new OutputStreamWriter(
				fs.create(pathOutputNode, true)));
		if (sb.toString().length() > 1)
			bwOut.write(sb.toString());
		bwOut.close();

		Path pathInputNode = new Path(Initial.PATH_INPUT_NODE_CACHE + "/"
				+ mapId);
		BufferedWriter bwIn = new BufferedWriter(new OutputStreamWriter(
				fs.create(pathInputNode, true)));
		if (sbIn.toString().length() > 1)
			bwIn.write(sbIn.toString());
		bwIn.close();
		long endTime3 = System.currentTimeMillis();

		context.write(NullWritable.get(), values);

		long endTime4 = System.currentTimeMillis();

		System.out.println((endTime1 - startTime1) + " --- "
				+ (endTime2 - endTime1) + " --- " + (endTime3 - endTime2)
				+ " --- " + (endTime4 - endTime3));

	}
	
	public static HashSet<Integer> listMarked;

	/**
	 * @param vfi
	 *            : a set of vertices in fragment i
	 * @param inodes
	 *            : a set of input node
	 * @param onodes
	 *            : a set of output node
	 * @param s
	 *            : start node in query
	 * @param t
	 *            : terminal node in query
	 * @param gq
	 *            : query automaton
	 * @return a set of vectors of the boolean formula
	 */
	public static HashSet<Rvec> localEval(HashMap<String, Vertex> vfi,
			HashMap<String, Vertex> inodes, HashMap<String, Vertex> onodes,
			GraphQuery gq) {

		HashSet<Rvec> rvset = new HashSet<Rvec>();

		ArrayList<VertexQ> Vq = gq.Vq; // get all vertex in query
		VertexQ ut = gq.ut; // get terminal state

		Vertex s = new Vertex(gq.us.stateName); // get start vertex
		Vertex t = new Vertex(gq.ut.stateName); // get terminal vertex

		HashMap<String, Vertex> iset = inodes; // set in-nodes
		HashMap<String, Vertex> oset = onodes; // set virtual nodes

		if (vfi.containsValue(t)) {
			if (!oset.containsValue(t)) {
				vfi.get(t.name).label = t.name;
				t = vfi.get(t.name);
				oset.put(t.name, t);
			} else {
				oset.get(t.name).label = t.name;
				vfi.get(t.name).label = t.name;
			}
		}

		// for(Vertex vt: iset){
		// if(vt.name.equals("1775"))
		// for(Edge ed: vt.neighbors)
		// System.out.print(ed.target+"\t");
		// }

		if (vfi.containsValue(s) && !oset.containsValue(s)) {
			if (!iset.containsValue(s)) {
				vfi.get(s.name).label = s.name;
				s = vfi.get(s.name);
				iset.put(s.name, s);
			} else {
				iset.get(s.name).label = s.name;
				vfi.get(s.name).label = s.name;
			}
		}

		// for (Vertex v : vfi)
		// v.visit = false;

		for (Vertex v : oset.values()) {
			v.rvec = new ArrayList<Rvec>(); // set empty

			// set value for v where indicate v can match of state u
			for (VertexQ u : Vq) {
				// v is terminal node
				if (v.equals(t) && u.equals(ut)) {
					v.rvec.add(new Rvec(v.name, u.stateName,
							new ArrayList<BoolFormula>(Arrays
									.asList(new BoolFormula(true)))));
					break;
				}
				// v is match of state u
				else if (v.L().equals(u.Lq())
						|| (!v.equals(t) && u.Lq().equals("_"))) // add _
				// case
				{
					v.rvec.add(new Rvec(v.name, u.stateName,
							new ArrayList<BoolFormula>(Arrays
									.asList(new BoolFormula(new X(v.name,
											u.stateName))))));
					break;
				}
			}

			v.visit = true;
		}

		// compute vector boolean formula for each input node
		for (Vertex v : iset.values()) {
			listMarked = new HashSet<Integer>();
			v.rvec = cmpRvec(v, vfi, iset, gq);
			if (v.rvec.size() > 0)
				rvset.addAll(v.rvec);
		}
		return rvset;
	}

	/**
	 * @param v
	 *            : input node
	 * @param vfi
	 *            : list of nodes in V
	 * @param gq
	 *            : query automaton
	 * @return: a vector Rvec of boolean formula
	 */
	public static ArrayList<Rvec> cmpRvec(Vertex v,
			HashMap<String, Vertex> vfi, HashMap<String, Vertex> iset,
			GraphQuery gq) {

		ArrayList<VertexQ> Vq = gq.Vq; // get all vertex in query

		if (v.visit == true)
			return v.rvec;

		listMarked.add(Integer.parseInt(v.name));

		// initial false for all state by v
		// for (VertexQ u : Vq)
		// if (v.L().equals(u.Lq()) || u.Lq().equals("_"))
		// v.rvec.add(new Rvec(v.name, u.stateName,
		// new ArrayList<BoolFormula>(Arrays
		// .asList(new BoolFormula(false)))));

		// get list children node of v
		HashSet<Vertex> childrenOfV = new HashSet<Vertex>();
		for (Edge e : v.neighbors)
			if (!listMarked.contains(Integer.parseInt(e.target))) {
				// Vertex child = getVertex(e.target, vfi);
				// if (child != null)
				// childrenOfV.add(child);
				//
				Vertex c = vfi.get(e.target);

				if (c != null)
					childrenOfV.add(c);
			}

		// if v do not child, set visited
		if (childrenOfV.size() == 0)
			v.visit = true;
		else {
			// recurse visit children node and compute vector Rvec via them
			for (Vertex w : childrenOfV) {
				if (w.visit == false)
					w.rvec = cmpRvec(w, vfi, iset, gq);

				for (VertexQ u : Vq) {
					if (v.L().equals(u.Lq()) || u.Lq().equals("_")) {

						if (v.rvec.size() > 0) {

							ArrayList<BoolFormula> list = v.rvec.get(0).value;

							if (list.get(0).value == false) {

								// compose value from children node w
								// where w match of children state of u
								ArrayList<BoolFormula> value = cmposeVec(u, w,
										w.rvec, gq);

								if (!value.isEmpty()) {
									for (BoolFormula bf : value)
										if (bf.boolVar != null) {
											if (!list.contains(bf))
												list.add(bf);
										} else if (bf.value == true) {
											list.clear();
											list.add(bf);
											break;
										}

									// if (list.size() > 0)
									// v.rvec.set(0, new Rvec(v.name,
									// u.stateName, list));

									if (list.size() > 0)
										v.rvec.set(0, new Rvec(v.name,
												u.stateName, list));

								}
							}

						} else {
							ArrayList<BoolFormula> list = new ArrayList<BoolFormula>();
							// NEW IMPROVE

							ArrayList<BoolFormula> value = cmposeVec(u, w,
									w.rvec, gq);

							if (!value.isEmpty()) {
								for (BoolFormula bf : value)
									if (bf.boolVar != null) {
										if (!list.contains(bf))
											list.add(bf);
									} else if (bf.value == true) {
										list.clear();
										list.add(bf);
										break;
									}
								if (list.size() > 0) {
									v.rvec.add(new Rvec(v.name, u.stateName,
											list));
								}

							}
						}
						break; // only matches at one state
					}

				}
			}
			v.visit = true;
		}

		return v.rvec;
	}

	/**
	 * @param u
	 *            : a state in Vq
	 * @param w
	 *            : a vertex in Vi
	 * @param wRvec
	 *            : a vector of Boolean formula of node w
	 * @return list of boolean formula
	 */
	public static ArrayList<BoolFormula> cmposeVec(VertexQ u, Vertex w,
			ArrayList<Rvec> wRvec, GraphQuery gq) {

		ArrayList<VertexQ> childrenOfU = new ArrayList<VertexQ>();
		ArrayList<BoolFormula> list = new ArrayList<BoolFormula>();

		// Get children state of u
		for (EdgeQ e : u.childrenState) {
			VertexQ child = getVertexQ(e.target, gq.Vq);
			childrenOfU.add(child);
		}

		// compose value
		for (VertexQ v : childrenOfU)
			if (w.L().equals(v.Lq()) || v.Lq().equals("_")) {
				for (Rvec r : wRvec)
					if (r.u.equals(v.Lq())) {
						list = r.value;
						break;
					}
				break;
			}
		return list;
	}

	public static void LoadDataDBLP(Text value,
			HashMap<String, Vertex> listViNode,
			HashMap<String, Vertex> listInNode,
			HashMap<String, Vertex> listOutNode) {
		try {
			byte[] bytes = value.getBytes();
			ByteArrayInputStream arr = new ByteArrayInputStream(bytes);
			BufferedReader bNode = new BufferedReader(
					new InputStreamReader(arr));

			String strLine = null;
			while ((strLine = bNode.readLine()) != null) {
				String[] row = strLine.split("\t");
				String[] node = row[0].split(":"); // split to get node id &
													// determine input node
													// in
													// fragment
				if (node.length >= 2) {
					Vertex vi;
					if (row.length >= 4)
						vi = new Vertex(node[0], row[4]);
					else
						vi = new Vertex(node[0]);

					for (int i = 6; i < row.length; i++) {
						vi.neighbors.add(new Edge(row[i]));
					}

					listViNode.put(vi.name, vi);
					if (node[1].equals("1"))
						listInNode.put(vi.name, vi);
					else if (node[1].equals("2"))
						listOutNode.put(vi.name, vi);
				}
			}

			bNode.close();

		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * @param filePath
	 */
	public static void LoadDataYouTube(Text value,
			HashMap<String, Vertex> listViNode,
			HashMap<String, Vertex> listInNode,
			HashMap<String, Vertex> listOutNode) {
		try {
			byte[] bytes = value.getBytes();
			ByteArrayInputStream arr = new ByteArrayInputStream(bytes);
			BufferedReader bNode = new BufferedReader(
					new InputStreamReader(arr));

			String strLine = null;
			while ((strLine = bNode.readLine()) != null) {
				String[] row = strLine.split("\t");
				String[] node = row[0].split(":"); // split to get node id &
													// determine input node in
													// fragment
				if (node.length >= 2) {
					Vertex vi;
					if (row.length >= 3)
						vi = new Vertex(node[0], row[3]);
					else
						vi = new Vertex(node[0]);

					for (int i = 9; i < row.length; i++) {
						vi.neighbors.add(new Edge(row[i]));
					}

					listViNode.put(vi.name, vi);
					if (node[1].equals("1"))
						listInNode.put(vi.name, vi);
					else if (node[1].equals("2"))
						listOutNode.put(vi.name, vi);
				}
			}
			bNode.close();

		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static void LoadDataQuyet(Text value,
			HashMap<String, Vertex> listViNode,
			HashMap<String, Vertex> listInNode,
			HashMap<String, Vertex> listOutNode) {
		try {
			byte[] bytes = value.getBytes();
			ByteArrayInputStream arr = new ByteArrayInputStream(bytes);
			BufferedReader bNode = new BufferedReader(
					new InputStreamReader(arr));

			String strLine = null;
			while ((strLine = bNode.readLine()) != null) {
				String[] row = strLine.split("\t");
				String[] node = row[0].split(":"); // split to get node id &
													// determine input node in
													// fragment
				Vertex vi;
				if (row.length >= 2)
					vi = new Vertex(node[0], row[1]);
				else
					vi = new Vertex(node[0]);

				for (int i = 2; i < row.length; i++) {
					vi.neighbors.add(new Edge(row[i]));
				}

				listViNode.put(vi.name, vi);
				if (node[1].equals("1"))
					listInNode.put(vi.name, vi);
				else if (node[1].equals("2"))
					listOutNode.put(vi.name, vi);
			}
			bNode.close();

		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static void LoadDataTung(Text value,
			HashMap<String, Vertex> listViNode,
			HashMap<String, Vertex> listInNode,
			HashMap<String, Vertex> listOutNode) {
		try {

			byte[] bytes = value.getBytes();
			ByteArrayInputStream arr = new ByteArrayInputStream(bytes);
			BufferedReader bNode = new BufferedReader(
					new InputStreamReader(arr));

			String strLine = null;
			while ((strLine = bNode.readLine()) != null) {
				String[] row = strLine.split(";");
				String[] node = row[0].split(":");
				Vertex vi;
				if (row.length >= 2) {
					if (!node[1].equals("0")) {
						String[] atrs = row[1].split("\t");
						if (atrs.length >= 2)
							vi = new Vertex(node[0], atrs[2]);
						else
							vi = new Vertex(node[0]);
					} else
						vi = new Vertex(node[0], row[1]);

				} else
					vi = new Vertex(node[0]);

				for (int i = 2; i < row.length; i++) {
					String[] cells = row[i].split(":");
					cells[0] = cells[0].replace("_", "");
					vi.neighbors.add(new Edge(cells[0]));
				}

				listViNode.put(vi.name, vi);
				if (node[1].equals("1"))
					listInNode.put(vi.name, vi);
				else if (node[1].equals("0"))
					listOutNode.put(vi.name, vi);

			}
			bNode.close();

		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static VertexQ getVertexQ(String name, ArrayList<VertexQ> Vq) {
		VertexQ tmp = null;
		for (VertexQ v : Vq)
			if (v.stateName.equals(name)) {
				tmp = v;
				break;
			}
		return tmp;
	}

	public static Rvec GetRvec(String u, Vertex v) {
		Rvec tmp = null;
		for (Rvec r : v.rvec)
			if (r.u.equals(u))
				return r;
		return tmp;
	}

	public static int GetIndexRvec(String u, Vertex v) {
		for (int i = 0; i < v.rvec.size(); i++)
			if (v.rvec.get(i).u.equals(u))
				return i;
		return -1;
	}

}
